use async_trait::async_trait;
use fluviox_shared::{Result, FluvioXError, models::DataRecord};
use crate::traits::{SmartModule, Filter, SmartModuleType, ProcessResult};
use serde_json::Value;
use tracing::{info, warn};

/// 条件过滤器
/// 基于字段值和条件表达式过滤记录
pub struct ConditionFilter {
    name: String,
    config: Option<ConditionFilterConfig>,
}

/// 条件过滤器配置
#[derive(Debug, Clone)]
pub struct ConditionFilterConfig {
    /// 过滤条件列表
    pub conditions: Vec<FilterCondition>,
    /// 条件组合逻辑（AND/OR）
    pub logic: LogicOperator,
}

/// 过滤条件
#[derive(Debug, Clone)]
pub struct FilterCondition {
    /// 字段路径（支持嵌套，如 "user.name"）
    pub field_path: String,
    /// 操作符
    pub operator: ComparisonOperator,
    /// 比较值
    pub value: Value,
    /// 是否忽略大小写（仅对字符串有效）
    pub ignore_case: bool,
}

/// 逻辑操作符
#[derive(Debug, Clone)]
pub enum LogicOperator {
    And,
    Or,
}

/// 比较操作符
#[derive(Debug, Clone)]
pub enum ComparisonOperator {
    /// 等于
    Equal,
    /// 不等于
    NotEqual,
    /// 大于
    GreaterThan,
    /// 大于等于
    GreaterThanOrEqual,
    /// 小于
    LessThan,
    /// 小于等于
    LessThanOrEqual,
    /// 包含（字符串）
    Contains,
    /// 不包含（字符串）
    NotContains,
    /// 以...开始
    StartsWith,
    /// 以...结束
    EndsWith,
    /// 在列表中
    In,
    /// 不在列表中
    NotIn,
    /// 存在（字段存在）
    Exists,
    /// 不存在（字段不存在）
    NotExists,
}

impl ConditionFilter {
    pub fn new() -> Self {
        Self {
            name: "condition-filter".to_string(),
            config: None,
        }
    }
    
    /// 获取嵌套字段值
    fn get_field_value<'a>(&self, data: &'a Value, field_path: &str) -> Option<&'a Value> {
        let parts: Vec<&str> = field_path.split('.').collect();
        let mut current = data;
        
        for part in parts {
            match current {
                Value::Object(obj) => {
                    current = obj.get(part)?;
                },
                _ => return None,
            }
        }
        
        Some(current)
    }
    
    /// 比较两个值
    fn compare_values(&self, left: &Value, operator: &ComparisonOperator, right: &Value, ignore_case: bool) -> bool {
        match operator {
            ComparisonOperator::Equal => {
                if ignore_case && left.is_string() && right.is_string() {
                    left.as_str().unwrap().to_lowercase() == right.as_str().unwrap().to_lowercase()
                } else {
                    left == right
                }
            },
            ComparisonOperator::NotEqual => {
                if ignore_case && left.is_string() && right.is_string() {
                    left.as_str().unwrap().to_lowercase() != right.as_str().unwrap().to_lowercase()
                } else {
                    left != right
                }
            },
            ComparisonOperator::GreaterThan => {
                self.numeric_compare(left, right, |a, b| a > b)
            },
            ComparisonOperator::GreaterThanOrEqual => {
                self.numeric_compare(left, right, |a, b| a >= b)
            },
            ComparisonOperator::LessThan => {
                self.numeric_compare(left, right, |a, b| a < b)
            },
            ComparisonOperator::LessThanOrEqual => {
                self.numeric_compare(left, right, |a, b| a <= b)
            },
            ComparisonOperator::Contains => {
                if let (Some(left_str), Some(right_str)) = (left.as_str(), right.as_str()) {
                    if ignore_case {
                        left_str.to_lowercase().contains(&right_str.to_lowercase())
                    } else {
                        left_str.contains(right_str)
                    }
                } else {
                    false
                }
            },
            ComparisonOperator::NotContains => {
                if let (Some(left_str), Some(right_str)) = (left.as_str(), right.as_str()) {
                    if ignore_case {
                        !left_str.to_lowercase().contains(&right_str.to_lowercase())
                    } else {
                        !left_str.contains(right_str)
                    }
                } else {
                    true
                }
            },
            ComparisonOperator::StartsWith => {
                if let (Some(left_str), Some(right_str)) = (left.as_str(), right.as_str()) {
                    if ignore_case {
                        left_str.to_lowercase().starts_with(&right_str.to_lowercase())
                    } else {
                        left_str.starts_with(right_str)
                    }
                } else {
                    false
                }
            },
            ComparisonOperator::EndsWith => {
                if let (Some(left_str), Some(right_str)) = (left.as_str(), right.as_str()) {
                    if ignore_case {
                        left_str.to_lowercase().ends_with(&right_str.to_lowercase())
                    } else {
                        left_str.ends_with(right_str)
                    }
                } else {
                    false
                }
            },
            ComparisonOperator::In => {
                if let Some(array) = right.as_array() {
                    array.contains(left)
                } else {
                    false
                }
            },
            ComparisonOperator::NotIn => {
                if let Some(array) = right.as_array() {
                    !array.contains(left)
                } else {
                    true
                }
            },
            ComparisonOperator::Exists => true,  // 如果能获取到值，说明字段存在
            ComparisonOperator::NotExists => false, // 如果能获取到值，说明字段存在
        }
    }
    
    /// 数值比较
    fn numeric_compare<F>(&self, left: &Value, right: &Value, compare_fn: F) -> bool
    where
        F: Fn(f64, f64) -> bool,
    {
        match (left.as_f64(), right.as_f64()) {
            (Some(l), Some(r)) => compare_fn(l, r),
            _ => false,
        }
    }
    
    /// 评估单个条件
    fn evaluate_condition(&self, data: &Value, condition: &FilterCondition) -> bool {
        match self.get_field_value(data, &condition.field_path) {
            Some(field_value) => {
                self.compare_values(field_value, &condition.operator, &condition.value, condition.ignore_case)
            },
            None => {
                // 字段不存在
                matches!(condition.operator, ComparisonOperator::NotExists)
            }
        }
    }
    
    /// 评估所有条件
    fn evaluate_conditions(&self, data: &Value, config: &ConditionFilterConfig) -> bool {
        if config.conditions.is_empty() {
            return true; // 没有条件，默认通过
        }
        
        match config.logic {
            LogicOperator::And => {
                config.conditions.iter().all(|condition| self.evaluate_condition(data, condition))
            },
            LogicOperator::Or => {
                config.conditions.iter().any(|condition| self.evaluate_condition(data, condition))
            },
        }
    }
}

#[async_trait]
impl SmartModule for ConditionFilter {
    fn name(&self) -> &str {
        &self.name
    }
    
    fn version(&self) -> &str {
        "1.0.0"
    }
    
    fn description(&self) -> &str {
        "条件过滤器，支持多种比较操作符和逻辑组合"
    }
    
    fn module_type(&self) -> SmartModuleType {
        SmartModuleType::Filter
    }
    
    async fn initialize(&mut self, config: Value) -> Result<()> {
        info!("初始化条件过滤器");
        
        // 解析条件列表
        let conditions = config.get("conditions")
            .and_then(|v| v.as_array())
            .map(|arr| {
                arr.iter()
                    .filter_map(|item| {
                        if let Value::Object(condition_obj) = item {
                            let field_path = condition_obj.get("field_path")?.as_str()?.to_string();
                            let operator_str = condition_obj.get("operator")?.as_str()?;
                            let value = condition_obj.get("value")?.clone();
                            let ignore_case = condition_obj.get("ignore_case")
                                .and_then(|v| v.as_bool())
                                .unwrap_or(false);
                            
                            let operator = match operator_str {
                                "eq" | "equal" => ComparisonOperator::Equal,
                                "ne" | "not_equal" => ComparisonOperator::NotEqual,
                                "gt" | "greater_than" => ComparisonOperator::GreaterThan,
                                "gte" | "greater_than_or_equal" => ComparisonOperator::GreaterThanOrEqual,
                                "lt" | "less_than" => ComparisonOperator::LessThan,
                                "lte" | "less_than_or_equal" => ComparisonOperator::LessThanOrEqual,
                                "contains" => ComparisonOperator::Contains,
                                "not_contains" => ComparisonOperator::NotContains,
                                "starts_with" => ComparisonOperator::StartsWith,
                                "ends_with" => ComparisonOperator::EndsWith,
                                "in" => ComparisonOperator::In,
                                "not_in" => ComparisonOperator::NotIn,
                                "exists" => ComparisonOperator::Exists,
                                "not_exists" => ComparisonOperator::NotExists,
                                _ => return None,
                            };
                            
                            Some(FilterCondition {
                                field_path,
                                operator,
                                value,
                                ignore_case,
                            })
                        } else {
                            None
                        }
                    })
                    .collect()
            })
            .unwrap_or_default();
        
        let logic = config.get("logic")
            .and_then(|v| v.as_str())
            .map(|s| match s.to_lowercase().as_str() {
                "or" => LogicOperator::Or,
                _ => LogicOperator::And,
            })
            .unwrap_or(LogicOperator::And);
        
        self.config = Some(ConditionFilterConfig {
            conditions,
            logic,
        });
        
        info!("条件过滤器初始化完成");
        Ok(())
    }
    
    async fn process(&mut self, record: DataRecord) -> Result<Vec<ProcessResult>> {
        let config = self.config.as_ref()
            .ok_or_else(|| FluvioXError::Internal("过滤器未初始化".to_string()))?;
        
        if self.evaluate_conditions(&record.data, config) {
            Ok(vec![ProcessResult::Keep(record)])
        } else {
            Ok(vec![ProcessResult::Filter])
        }
    }
}

#[async_trait]
impl Filter for ConditionFilter {
    async fn should_keep(&self, record: &DataRecord) -> Result<bool> {
        let config = self.config.as_ref()
            .ok_or_else(|| FluvioXError::Internal("过滤器未初始化".to_string()))?;
        
        Ok(self.evaluate_conditions(&record.data, config))
    }
}

impl Default for ConditionFilter {
    fn default() -> Self {
        Self::new()
    }
}
